package mysql;

import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.io.jdbc.JDBCInputFormat;
import org.apache.flink.api.java.io.jdbc.JDBCOutputFormat;
import org.apache.flink.api.java.typeutils.RowTypeInfo;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableEnvironment;
import org.apache.flink.table.api.java.BatchTableEnvironment;
import org.apache.flink.types.Row;

public class OrcaleTable
{
    public static void main(String[] args) throws Exception
    {
        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);
        BatchTableEnvironment tEnv  = TableEnvironment.getTableEnvironment(env);
        DataSet<Row> dbData =
        env.createInput(
                JDBCInputFormat.buildJDBCInputFormat()
                        .setDrivername(OracleConfigKeys.DRIVER_CLASS)
                        .setDBUrl(OracleConfigKeys.SOURCE_DRIVER_URL)
                        .setUsername(OracleConfigKeys.SOURCE_USER)
                        .setPassword(OracleConfigKeys.SOURCE_PASSWORD)
                        .setQuery("SELECT fee,'201812' stat_month,'CBSS' data_source,user_id from cbss.ts_b_bill_his PARTITION (\"P_201812\") WHERE cycle_id=201812 ")
                        .setRowTypeInfo(new RowTypeInfo(BasicTypeInfo.BIG_DEC_TYPE_INFO, BasicTypeInfo.STRING_TYPE_INFO,BasicTypeInfo.STRING_TYPE_INFO,BasicTypeInfo.BIG_DEC_TYPE_INFO))
                        .finish()
        );

        DataSet<Row> dbData1 =
                env.createInput(
                        JDBCInputFormat.buildJDBCInputFormat()
                                .setDrivername(OracleConfigKeys.DRIVER_CLASS)
                                .setDBUrl(OracleConfigKeys.SOURCE_DRIVER_URL)
                                .setUsername(OracleConfigKeys.SOURCE_USER)
                                .setPassword(OracleConfigKeys.SOURCE_PASSWORD)
                                .setQuery("SELECT 12 date_id,user_id, in_depart_id FROM cbss.TF_F_USER_his PARTITION (\"P_201812\") ")
                                .setRowTypeInfo(new RowTypeInfo(BasicTypeInfo.BIG_DEC_TYPE_INFO,BasicTypeInfo.BIG_DEC_TYPE_INFO, BasicTypeInfo.STRING_TYPE_INFO))
                                .finish()
                );

        tEnv.registerDataSet("ts_b_bill_his", dbData, "fee,stat_month,data_source,user_id");
        tEnv.registerDataSet("TF_F_USER_his", dbData1, "date_id,user_id,in_depart_id");

        Table table = tEnv.sqlQuery(
                "SELECT TF_F_USER_his.date_id AS date_id ,ts_b_bill_his.stat_month AS stat_month, ts_b_bill_his.data_source AS data_source, TF_F_USER_his.in_depart_id AS in_depart_id, SUM(ts_b_bill_his.fee) AS fee FROM ts_b_bill_his INNER JOIN TF_F_USER_his ON ts_b_bill_his.user_id = TF_F_USER_his.user_id GROUP BY TF_F_USER_his.date_id,TF_F_USER_his.in_depart_id ,ts_b_bill_his.stat_month, ts_b_bill_his.data_source");


        DataSet<Row> result = tEnv.toDataSet(table, new RowTypeInfo(BasicTypeInfo.BIG_DEC_TYPE_INFO,BasicTypeInfo.STRING_TYPE_INFO,
                BasicTypeInfo.STRING_TYPE_INFO,BasicTypeInfo.STRING_TYPE_INFO,BasicTypeInfo.BIG_DEC_TYPE_INFO));

        result.output(
                JDBCOutputFormat.buildJDBCOutputFormat()
                        .setDrivername(OracleConfigKeys.DRIVER_CLASS1)
                        .setDBUrl(OracleConfigKeys.SOURCE_DRIVER_URL1)
                        .setUsername(OracleConfigKeys.SOURCE_USER1)
                        .setPassword(OracleConfigKeys.SOURCE_PASSWORD1)
                        .setQuery("INSERT INTO ucr_data1.TS_s_bill(date_id, stat_month ,data_source ,in_depart_id, fee) values(?,?,?,?,?)")
                        .finish()
        );



        env.execute("flink mysql Dataset demo");
    }

}
